package test

import (
	"6.824/mr"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"
	"unicode"
)

func Map() {
	fileName := "D:\\code\\gitee\\6.824\\src\\main\\pg-being_ernest.txt"
	content, err := ioutil.ReadFile(fileName)
	if err != nil {
		fmt.Println("read file error")
	}
	kva := earlyMap(fileName, string(content))
	NReduce := 10
	intermediate := make([][]mr.KeyValue, NReduce)
	for _, kv := range kva {
		idx := ihash(kv.Key) % NReduce
		intermediate[idx] = append(intermediate[idx], kv)
	}
	for index, v := range intermediate {
		filename := reduceName(3, index)
		f, err := ioutil.TempFile("", filename+"_tmp")
		if err != nil {
			fmt.Println("TempFile file error")
		}
		enc := json.NewEncoder(f)
		for _, kv := range v {
			enc.Encode(&kv)
		}
		log.Println(f.Name())
		f.Close()
		err2 := os.Rename(f.Name(), filename)
		if nil != err2 {
			log.Fatalf("map rename error %s  %+v", filename, err2)
		}
	}
	log.Println("doMapTask success -- " + fileName)
}

func reduce2() {
	maps := make(map[string][]string)
	for j := 0; j < 9; j++ {
		for i := 0; i < 9; i++ {
			filename := "C:\\Users\\lihong\\Desktop\\new\\t\\" + reduceName(j, i)
			file, err := os.Open(filename)
			if nil != err {
				continue
			}
			dec := json.NewDecoder(file)
			for {
				var kv mr.KeyValue
				if err := dec.Decode(&kv); err != nil {
					break
				}
				if _, ok := maps[kv.Key]; !ok {
					maps[kv.Key] = make([]string, 0, 100)
				}
				maps[kv.Key] = append(maps[kv.Key], kv.Value)
			}
		}
	}
	for k, v := range maps {
		fmt.Printf("%v %v\n", k, earlyReduce(k, v))
	}

}

func Reduce(key string, values []string) string {
	// sort values to ensure deterministic output.
	vv := make([]string, len(values))
	copy(vv, values)
	sort.Strings(vv)

	val := strings.Join(vv, " ")
	return val
}

func reduceName(mapIdx, reduceIdx int) string {
	return fmt.Sprintf("mr-%d-%d", mapIdx, reduceIdx)
}

func wcMap(filename string, contents string) []mr.KeyValue {
	// function to detect word separators.
	ff := func(r rune) bool { return !unicode.IsLetter(r) }

	// split contents into an array of words.
	words := strings.FieldsFunc(contents, ff)

	kva := []mr.KeyValue{}
	for _, w := range words {
		kv := mr.KeyValue{w, "1"}
		kva = append(kva, kv)
	}
	return kva
}

func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

func earlyMap(filename string, contents string) []mr.KeyValue {
	kva := []mr.KeyValue{}
	kva = append(kva, mr.KeyValue{filename, "1"})
	return kva
}

func earlyReduce(key string, values []string) string {
	// some reduce tasks sleep for a long time; potentially seeing if
	// a worker will accidentally exit early
	if strings.Contains(key, "sherlock") || strings.Contains(key, "tom") {
		time.Sleep(time.Duration(3 * time.Second))
	}
	// return the number of occurrences of this file.
	return strconv.Itoa(len(values))
}
